/**
 * MemCached Java client
 * Copyright (c) 2007 Greg Whalin
 * All rights reserved.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the BSD license
 *
 * This library is distributed in the hope that it will be
 * useful, but WITHOUT ANY WARRANTY; without even the implied
 * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
 * PURPOSE.
 *
 * You should have received a copy of the BSD License along with this
 * library.
 *
 * @author Greg Whalin <greg@meetup.com> 
 * @version 2.0
 */
package stc.skymobi.netty.handler.codec.memcached;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageCodec;
import stc.skymobi.bean.bytebean.ByteBean;
import stc.skymobi.bean.bytebean.annotation.ByteField;
import stc.skymobi.bean.bytebean.codec.*;
import stc.skymobi.bean.bytebean.core.*;
import stc.skymobi.util.ByteOrder;
import stc.skymobi.util.DefaultNumberCodecs;

import java.lang.reflect.Field;
import java.util.List;

/**
 * Handle encoding standard Java types directly which can result in significant
 * memory savings:
 * <p/>
 * Currently the Memcached driver for Java supports the setSerialize() option.
 * This can increase performance in some situations but has a few issues:
 * <p/>
 * Code that performs class casting will throw ClassCastExceptions when
 * setSerialize is enabled. For example:
 * <p/>
 * mc.set( "foo", new Integer( 1 ) ); Integer output = (Integer)mc.get("foo");
 * <p/>
 * Will work just file when setSerialize is true but when its false will just
 * throw a ClassCastException.
 * <p/>
 * Also internally it doesn't support Boolean and since toString is called
 * wastes a lot of memory and causes additional performance issue. For example
 * an Integer can take anywhere from 1 byte to 10 bytes.
 * <p/>
 * Due to the way the memcached slab allocator works it seems like a LOT of
 * wasted memory to store primitive types as serialized objects (from a
 * performance and memory perspective). In our applications we have millions of
 * small objects and wasted memory would become a big problem.
 * <p/>
 * For example a Serialized Boolean takes 47 bytes which means it will fit into
 * the 64byte LRU. Using 1 byte means it will fit into the 8 byte LRU thus
 * saving 8x the memory. This also saves the CPU performance since we don't have
 * to serialize bytes back and forth and we can compute the byte[] value
 * directly.
 * <p/>
 * One problem would be when the user calls get() because doing so would require
 * the app to know the type of the object stored as a bytearray inside memcached
 * (since the user will probably cast).
 * <p/>
 * If we assume the basic types are interned we could use the first byte as the
 * type with the remaining bytes as the value. Then on get() we could read the
 * first byte to determine the type and then construct the correct object for
 * it. This would prevent the ClassCastException I talked about above.
 * <p/>
 * We could remove the setSerialize() option and just assume that standard VM
 * types are always internd in this manner.
 * <p/>
 * mc.set( "foo", new Boolean.TRUE ); Boolean b = (Boolean)mc.get( "foo" );
 * <p/>
 * And the type casts would work because internally we would create a new
 * Boolean to return back to the client.
 * <p/>
 * This would reduce memory footprint and allow for a virtual implementation of
 * the Externalizable interface which is much faster than Serialzation.
 * <p/>
 * Currently the memory improvements would be:
 * <p/>
 * java.lang.Boolean - 8x performance improvement (now just two bytes)
 * java.lang.Integer - 16x performance improvement (now just 5 bytes)
 * <p/>
 * Most of the other primitive types would benefit from this optimization.
 * java.lang.Character being another obvious example.
 * <p/>
 * I know it seems like I'm being really picky here but for our application I'd
 * save 1G of memory right off the bat. We'd go down from 1.152G of memory used
 * down to 144M of memory used which is much better IMO.
 * <p/>
 * http://java.sun.com/docs/books/tutorial/native1.1/integrating/types.html
 *
 * @author <a href="mailto:burton@peerfear.org">Kevin A. Burton</a>
 * @author Greg Whalin <greg@meetup.com>
 */
public class MemcachedNativeCodec extends ByteToMessageCodec {
    private static final Field2Desc field2Desc;
    private static final DefaultCodecProvider codecProvider;
    private static final ByteOrder BYTE_ORDER = ByteOrder.BigEndian;

    private MemcachedRequestEncoder encoder = new MemcachedRequestEncoder();
    //lengthAdjustment =  the length of HDR1 + LEN, negative,we need contains header,so plus the length of header
    private MemcachedResponseDecoder decoder = new MemcachedResponseDecoder(Integer.MAX_VALUE, 8, 4, -(8 + 4) + 24, 0);

    static {
        field2Desc = new Field2Desc() {

            public ByteFieldDesc genDesc(Field field) {
                ByteField byteField = field.getAnnotation(ByteField.class);
                Class<?> clazz = field.getDeclaringClass();
                if (null != byteField) {
                    try {
                        Field lengthField = null;
                        if (!byteField.length().equals("")) {
                            Class<?> cls = field.getDeclaringClass();
                            do {
                                try {
                                    lengthField = cls.getDeclaredField(byteField.length());
                                } catch (NoSuchFieldException e) {
                                    cls = cls.getSuperclass();
                                }
                            } while (cls != null && lengthField == null);
                        }
                        DefaultFieldDesc desc
                                = new DefaultFieldDesc()
                                .setField(field)
                                .setIndex(byteField.index())
                                .setByteSize(byteField.bytes())
                                .setCharset(byteField.charset())
                                .setLengthField(lengthField)
                                .setCustomTypeMethod(
                                        byteField.customType().equals("")
                                                ? null
                                                : clazz.getDeclaredMethod(byteField.customType()))
                                .setByteOrder(byteField.byteOrder())
                                .setFixedLength(byteField.fixedLength());
                        return desc;
                    } catch (SecurityException e) {
                        e.printStackTrace();
                    } catch (NoSuchMethodException e) {
                        e.printStackTrace();
                    }
                }
                return null;
            }
        };

        codecProvider = new DefaultCodecProvider()
                .addCodec(new MemcachedStringCodec())
                .addCodec(new LongCodec())
                .addCodec(new IntCodec())
                .addCodec(new ShortCodec())
                .addCodec(new ByteCodec())
                .addCodec(new ByteArrayCodec())
                .addCodec(new ArrayCodec())
                .addCodec(new AnyCodec(ByteBean.class));

        BeanCodec beanCodec = new BeanCodec(ByteBean.class, field2Desc);

        DefaultDecContextFactory decContextFactory = new DefaultDecContextFactory();
        decContextFactory.setCodecProvider(codecProvider);
        decContextFactory.setNumberCodec(
                DefaultNumberCodecs.byteOrder2NumberCodec(BYTE_ORDER));

        beanCodec.setDecContextFactory(decContextFactory);

        DefaultEncContextFactory encContextFactory = new DefaultEncContextFactory();
        encContextFactory.setCodecProvider(codecProvider);
        encContextFactory.setNumberCodec(
                DefaultNumberCodecs.byteOrder2NumberCodec(BYTE_ORDER));

        beanCodec.setEncContextFactory(encContextFactory);

        codecProvider.addCodec(beanCodec);
    }

    public static BeanFieldCodec getBeanCodec() {
        return (BeanFieldCodec) codecProvider.getCodecOf(FieldCodecCategory.BEAN);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out)
            throws Exception {
        out.add(decoder.decode(ctx, in));
    }

    @Override
    public void encode(ChannelHandlerContext ctx, Object in, ByteBuf out)
            throws Exception {
        encoder.encode(ctx, in, out);
    }
}
